package com.asiainfo.dacp.datastash.broker.flow.sharding.service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

import org.apache.commons.vfs2.FileObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import com.asiainfo.dacp.datastash.broker.core.BrokerContext;
import com.asiainfo.dacp.datastash.broker.core.BrokerException;
import com.asiainfo.dacp.datastash.broker.core.enums.ErrorCodeEnum;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitStepShardMeta;
import com.asiainfo.dacp.datastash.broker.flow.core.domain.Flow;
import com.asiainfo.dacp.datastash.domain.unit.models.DataStashUnitStep;
import com.asiainfo.dacp.metamodel.domain.ds.model.MetaDataSource;
import com.asiainfo.dacp.util.StringUtils;
import com.asiainfo.dacp.util.vfs.VfsUtils;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;

@Service
public class ShardingService {
	private static Logger LOG = LoggerFactory.getLogger(ShardingService.class);

	@SuppressWarnings("unchecked")
	public List<List<Map<String, String>>> sharding(DataStashUnitStep stepDomain, Flow flow) throws BrokerException {
		Map<String, Object> cfg = stepDomain.getStepCfgValMap();

		int nShardCnt = (Integer) cfg.get("shardCnt");
		List<Map<String, Object>> contents = (List<Map<String, Object>>) cfg.get("contents");
		String strg = (String) cfg.get("strg");

		List<List<Map<String, String>>> shardContents = Lists.newArrayList();

		if (contents == null || contents.size() == 0) {
			return shardContents;
		}

		List<StashUnitStepShardMeta> fileContents = new ArrayList<>();

		for (Map<String, Object> cont : contents) {
			String dsName = (String) cont.get("dsName");
			String contentStr = (String) cont.get("content");
			StashUnitStepShardMeta shardMeta = new StashUnitStepShardMeta(dsName, contentStr, strg);
			String filter = "";
			if (cont.get("filter") != null) {
				filter = StringUtils.variableSubstitution((String) cont.get("filter"), flow.getUnit());
			}
			shardMeta.setFilter(filter);
			fileContents.add(shardMeta);
		}

		List<Map<String, String>> elements = discovery(fileContents, flow);

		int eleCnt = elements.size() / nShardCnt + (elements.size() % nShardCnt == 0 ? 0 : 1);

		List<Map<String, String>> pieceMap = null;
		for (int i = 0; i < elements.size(); i++) {
			if (i % eleCnt == 0) {
				pieceMap = new ArrayList<>();
				shardContents.add(pieceMap);
			}
			pieceMap.add(elements.get(i));
		}

		return shardContents;
	}

	private List<Map<String, String>> discovery(List<StashUnitStepShardMeta> fileContents, Flow flow)
			throws BrokerException {

		List<Map<String, String>> result = new ArrayList<>();
		// TODO 需要实现多种数据文件的识别

		// 1.目录下面，或目录的的正则
		// 2.打开状态文件，按照分割
		for (StashUnitStepShardMeta shardMeta : fileContents) {

			if ("file".equals(shardMeta.getStrg()) || "hdfs".equals(shardMeta.getStrg())) {
				FileObject stateFilePath = null;
				try {
					MetaDataSource dataSource = BrokerContext.getInstance().getDsDefs().get(shardMeta.getDsName());

					if ("hdfs".equals(shardMeta.getStrg()) && !Strings.isNullOrEmpty(dataSource.getDsAcct())) {
						System.setProperty("HADOOP_USER_NAME", dataSource.getDsAcct());
					}

					String vfsPath = dataSource.getVfsHeader()
							+ StringUtils.variableSubstitution(shardMeta.getContent(), flow.getUnit());
					stateFilePath = VfsUtils.getInstance().resolveFile(vfsPath);

					if (!stateFilePath.exists()) {
						String errMsg = String.format("[%s] 配置路径 [%s] 不存在, 请确认.", flow.buildSerNo(), vfsPath);
						LOG.error(errMsg);
						throw new BrokerException(ErrorCodeEnum.CONFIG_ERROR, errMsg);
					}

					if (stateFilePath.isFile()) {
						result.add(getContent(shardMeta.getDsName(), stateFilePath.getName().getPath()));
					} else if (stateFilePath.isFolder()) {
						FileObject[] fileObjects = stateFilePath.getChildren();
						if (fileObjects != null && fileObjects.length > 0) {
							Pattern pattern = null;
							// TODO 需要实现目录下面的正则匹配
							if (org.apache.commons.lang3.StringUtils.isNotBlank(shardMeta.getFilter())) {
								pattern = Pattern.compile(shardMeta.getFilter());
							}
							for (FileObject file : fileObjects) {
								String pathName = file.getName().getPath();
								String fileName = file.getName().getBaseName();
								if (pattern == null || pattern.matcher(fileName).matches()) {
									result.add(getContent(shardMeta.getDsName(), pathName));
								}
							}
						}
					}
				} catch (Exception e) {
					String errMsg = String.format("[%s] 查找分片内容失败.", flow.buildSerNo(), e.getMessage());
					LOG.error(errMsg, e);
					throw new BrokerException(ErrorCodeEnum.CONFIG_ERROR, errMsg);
				} finally {
					VfsUtils.getInstance().closeFileObject(stateFilePath);
				}

			} else if ("sql".equals(shardMeta.getStrg())) {
				result.add(getSql(shardMeta.getDsName(), shardMeta.getContent()));
			}
		}
		LOG.info(String.format("[%s] 成功数据内容发现，分片数: %d ." ,flow.buildSerNo(), result.size()));
		return result;
	}

	private Map<String, String> getContent(String dsName, String contentStr) {
		Map<String, String> content = new HashMap<>();
		content.put("dsName", dsName);
		content.put("content", contentStr);
		return content;
	}

	private Map<String, String> getSql(String dsName, String sql) {
		Map<String, String> content = new HashMap<>();
		content.put("dsName", dsName);
		content.put("sql", sql);
		return content;
	}

}
